package broadcast

import (
	"encoding/hex"
	"fmt"
	"runtime"

	"hundsun.com/hsl/hschain/common/util"

	"github.com/golang/protobuf/proto"
	"github.com/golang/snappy"
	lru "github.com/hashicorp/golang-lru"
	logger "hundsun.com/hsl/hschain/common/log"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/p2p/impl/dht/node/pubsub"
	ltypes "hundsun.com/hsl/hschain/p2p/impl/dht/protocol/broadcast/types"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
	protosp2p "hundsun.com/hsl/hschain/protos/p2p"
	"hundsun.com/hsl/hschain/store/mq"
)

//V0 基于libp2p pubsub插件广播
type V0 struct {
	*Protocol
	Pubsub *pubsub.PubSub // libp2p广播插件
}

// NewBroadcastV0 NewPubSub
func NewBroadcastV0(b *Protocol) *V0 {
	// TODO topic后续可以加入区块链名以及通道来进行平行链的区分
	return &V0{Protocol: b}
}

//Init 广播入口函数，处理相关初始化
func (p *V0) Init() {
	// TODO opts后续有需求再实现
	p.Log.Info("broadcast v0 starting")
	pub, err := pubsub.NewPubSub(p.Ctx, p.P2pNode.Host, p.Log, p.P2pNode.Manager)
	if err != nil {
		panic(fmt.Sprintln("init broadcast protocol fail", err))
	}
	p.Pubsub = pub

	txIncoming := make(chan pubsub.SubMsg, 2048) //交易接收通道, 订阅外部广播消息
	txOutgoing := p.Comps.Sub(ltypes.TxTopic)    //交易发送通道, 订阅内部广播消息
	//区块
	blockIncoming := make(chan pubsub.SubMsg, 128)
	blockOutgoing := p.Comps.Sub(ltypes.BlockTopic)

	// pub sub topic注册
	err = p.Pubsub.JoinAndSubTopic(ltypes.TxTopic, p.callback(txIncoming))
	if err != nil {
		p.Log.Error("pubsub broadcast", "join tx topic err", err)
		return
	}
	err = p.Pubsub.JoinAndSubTopic(ltypes.BlockTopic, p.callback(blockIncoming))
	if err != nil {
		p.Log.Error("pubsub broadcast", "join block topic err", err)
		return
	}

	// 不存在订阅topic的节点时，不开启广播，目前只在初始化时做判定
	/*for len(p.Pubsub.FetchTopicPeers(ltypes.TxTopic)) == 0 {
		//todo 改为常量，重试n次之后打印日志
		time.Sleep(time.Second * 20)
		p.Log.Warn("pub sub broadcast", "info", "no peers available")
	}*/

	//发送和接收用多个函数并发处理，提高效率
	//交易广播, 使用多个协程并发处理，提高效率
	cpu := runtime.NumCPU()
	for i := 0; i < cpu; i++ {
		go p.handlePubMsg(ltypes.TxTopic, txOutgoing)
		go p.handleSubMsg(ltypes.TxTopic, txIncoming, p.TxFilter)
	}

	//区块广播
	go p.handlePubMsg(ltypes.BlockTopic, blockOutgoing)
	go p.handleSubMsg(ltypes.BlockTopic, blockIncoming, p.BlockFilter)
	p.Log.Info("broadcast v0 started")

}

// 处理广播消息发布
func (p *V0) handlePubMsg(topic string, out chan interface{}) {
	defer p.Comps.Unsub(out)
	buf := make([]byte, 0)
	var err error
	for {
		select {
		case data, ok := <-out: //发送广播交易
			if !ok {
				return
			}
			msg := data.(proto.Message)
			raw := p.encodeMsg(msg, &buf)
			if err != nil {
				p.Log.Error("handlePubMsg", "topic", topic, "hash", p.getMsgHash(topic, msg), "err", err)
				break
			}

			err = p.Pubsub.Publish(topic, raw)
			if err != nil {
				p.Log.Error("handlePubMsg", "topic", topic, "publish err", err)
			}

		case <-p.Ctx.Done():
			return
		}
	}
}

// 处理广播消息订阅
func (p *V0) handleSubMsg(topic string, in chan pubsub.SubMsg, filter *lru.Cache) {

	buf := make([]byte, 0)
	var err error
	for {
		select {
		case data, ok := <-in: //接收广播交易
			if !ok {
				return
			}
			p2pMsg := &protosp2p.P2PMsg{}
			err = p.decodeMsg(data.Data, &buf, p2pMsg)
			if err != nil {
				p.Log.Error("handleSubMsg", "topic", topic, "decodeMsg err", err)
				break
			}

			if topic == ltypes.TxTopic {
				var txs = &pbledger.Transactions{}
				_, err := util.Bytes2Struct(txs, p2pMsg.GetPayload())
				if err != nil {
					p.Log.Errorf("tx deserialize failed, reason:%v", err)
					break
				}

				//判断交易是否重复
				var txToInsert []*pbledger.Transaction
				for _, tx := range txs.Txs {
					hash := tx.GetXXXKey()
					if !filter.Contains(hash) {
						txToInsert = append(txToInsert, tx)
						filter.Add(tx.GetXXXKey(), struct{}{})
					}
				}
				if txToInsert == nil {
					break
				}

				err = p.TxPool.AddTransactions(txToInsert)
				if err != nil {
					logger.Errorf("tx pending failed reason:%v", err)
				}
				p.Log.Infof("receive txs from brother node, tx num:%d", len(txs.Txs))
			} else {
				var blk = &pbledger.Block{}
				_, err = util.Bytes2Struct(blk, p2pMsg.GetPayload())
				p.Log.Infof("receive broadcast block, blkNumber:%d", blk.Header.Number)

				if err != nil {
					logger.Errorf("block serialize failed, reason:%v", err)
					return
				}

				//判断区块是否重复
				hash := blk.GetXXXKey()
				if filter.Contains(hash) {
					break
				}
				filter.Add(hash, struct{}{})

				var blocks = make([]*pbledger.Block, 1)
				blocks = append(blocks, blk)
				msgBlock := p.TopicClient.NewMessage(types.BlockChainModule, mq.ModeTopic.String(), types.EventBroadcastAddBlock, blocks)
				if err := p.TopicClient.Send(msgBlock, false); err != nil {
					p.Log.Errorf("send broadcast add block failed reason:%v, blkNumber", err, blk.Header.Number)
				}
			}

			if err != nil {
				p.Log.Error("handleSubMsg", "topic", topic, "post msg err", err)
			}

		case <-p.Ctx.Done():
			return
		}
	}

}

// 统一处理哈希计算
func (p *V0) getMsgHash(topic string, msg proto.Message) string {
	if topic == ltypes.TxTopic {
		return hex.EncodeToString(msg.(*pbledger.Transaction).Hash())
	}
	return hex.EncodeToString(msg.(*pbledger.Block).Hash())
}

// 生成订阅消息回调
func (p *V0) callback(out chan<- pubsub.SubMsg) pubsub.SubCallBack {
	return func(topic string, msg pubsub.SubMsg) {
		out <- msg
	}
}

// 数据压缩后发送， 内部对相关数组进行重复利用
func (p *V0) encodeMsg(msg proto.Message, pbuf *[]byte) []byte {
	buf := *pbuf
	buf = buf[:cap(buf)]
	raw, err := proto.Marshal(msg)
	if err != nil {
		panic(err)
	}
	buf = snappy.Encode(buf, raw)
	*pbuf = buf
	// 复用raw数组作为压缩数据返回， 需要比较容量是否够大
	if cap(raw) >= len(buf) {
		raw = raw[:len(buf)]
	} else {
		raw = make([]byte, len(buf))
	}
	copy(raw, buf)
	return raw
}

// 接收数据并解压缩
func (p *V0) decodeMsg(raw []byte, pbuf *[]byte, msg proto.Message) error {
	var err error
	buf := *pbuf
	buf = buf[:cap(buf)]
	buf, err = snappy.Decode(buf, raw)
	if err != nil {
		p.Log.Error("pubSub decodeMsg", "snappy decode err", err)
		return err
	}
	//重复利用解码buf
	//*pbuf = buf
	err = proto.Unmarshal(buf, msg)
	if err != nil {
		p.Log.Error("pubSub decodeMsg", "pb decode err", err)
		return err
	}

	return nil
}
